/*-------------------------------------------------------------------------
 *
 * schemacmds.c
 *	  模式创建/操作命令
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  src/backend/commands/schemacmds.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/htup_details.h"
#include "access/table.h"
#include "access/xact.h"
#include "catalog/catalog.h"
#include "catalog/dependency.h"
#include "catalog/indexing.h"
#include "catalog/namespace.h"
#include "catalog/objectaccess.h"
#include "catalog/pg_authid.h"
#include "catalog/pg_namespace.h"
#include "commands/dbcommands.h"
#include "commands/event_trigger.h"
#include "commands/schemacmds.h"
#include "miscadmin.h"
#include "parser/parse_utilcmd.h"
#include "parser/scansup.h"
#include "tcop/utility.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/rel.h"
#include "utils/syscache.h"

static void fc_AlterSchemaOwner_internal(HeapTuple fc_tup, Relation fc_rel, Oid fc_newOwnerId);

/*
 * 创建模式
 *
 * 注意：调用者应传入整个
 * CREATE SCHEMA 语句的位置
 * 信息，而我们将其作为组件命令的位置传递。
 * 这符合我们的总体计划，即
 * 即使在执行子查询时，也报告整个命令的位置/长度。
 */
Oid CreateSchemaCommand(CreateSchemaStmt *fc_stmt, const char *fc_queryString,
					int fc_stmt_location, int fc_stmt_len)
{
	const char *fc_schemaName = fc_stmt->schemaname;
	Oid			fc_namespaceId;
	List	   *fc_parsetree_list;
	ListCell   *fc_parsetree_item;
	Oid			fc_owner_uid;
	Oid			fc_saved_uid;
	int			fc_save_sec_context;
	int			fc_save_nestlevel;
	char	   *fc_nsp = namespace_search_path;
	AclResult	fc_aclresult;
	ObjectAddress fc_address;
	StringInfoData fc_pathbuf;

	GetUserIdAndSecContext(&fc_saved_uid, &fc_save_sec_context);

	/*
	 * 谁应该拥有新模式？
	 */
	if (fc_stmt->authrole)
		fc_owner_uid = get_rolespec_oid(fc_stmt->authrole, false);
	else
		fc_owner_uid = fc_saved_uid;

	/* 如果未指定，则用用户名填充模式名称 */
	if (!fc_schemaName)
	{
		HeapTuple	fc_tuple;

		fc_tuple = SearchSysCache1(AUTHOID, ObjectIdGetDatum(fc_owner_uid));
		if (!HeapTupleIsValid(fc_tuple))
			elog(ERROR, "cache lookup failed for role %u", fc_owner_uid);
		fc_schemaName =
			pstrdup(NameStr(((Form_pg_authid) GETSTRUCT(fc_tuple))->rolname));
		ReleaseSysCache(fc_tuple);
	}

	/*
	 * 要创建模式，必须在当前
	 * 数据库上具有 schema-create 权限，并且必须能够成为目标角色（这并不
	 * 意味着目标角色本身必须具有创建模式的权限）。
	 * 后者条款防止 "赠送" 攻击。注意，超级用户将始终具有这两项权限。
	 */
	fc_aclresult = pg_database_aclcheck(MyDatabaseId, fc_saved_uid, ACL_CREATE);
	if (fc_aclresult != ACLCHECK_OK)
		aclcheck_error(fc_aclresult, OBJECT_DATABASE,
					   get_database_name(MyDatabaseId));

	check_is_member_of_role(fc_saved_uid, fc_owner_uid);

	/* 额外检查以保护保留的模式名称 */
	if (!allowSystemTableMods && IsReservedName(fc_schemaName))
		ereport(ERROR,
				(errcode(ERRCODE_RESERVED_NAME),
				 errmsg("unacceptable schema name \"%s\"", fc_schemaName),
				 errdetail("The prefix \"pg_\" is reserved for system schemas.")));

	/*
	 * 如果给定了 if_not_exists 且模式已存在，则退出。
	 * （注意：当不使用 if_not_exists 时我们不需要检查，因为
	 * NamespaceCreate 无论如何会抱怨。）我们可以在进行
	 * 权限检查之前进行此检查，但由于 CREATE TABLE IF NOT EXISTS 首先进行其
	 * 创建权限检查，我们也这样做。
	 */
	if (fc_stmt->if_not_exists)
	{
		fc_namespaceId = get_namespace_oid(fc_schemaName, true);
		if (OidIsValid(fc_namespaceId))
		{
			/*
			 * 如果我们处于扩展脚本中，必须确保预先存在的
			 * 对象是扩展的成员，以避免安全风险。
			 */
			ObjectAddressSet(fc_address, NamespaceRelationId, fc_namespaceId);
			checkMembershipInCurrentExtension(&fc_address);

			/* 可以跳过 */
			ereport(NOTICE,
					(errcode(ERRCODE_DUPLICATE_SCHEMA),
					 errmsg("schema \"%s\" already exists, skipping",
							fc_schemaName)));
			return InvalidOid;
		}
	}

	/*
	 * 如果请求的授权与当前用户不同，
	 * 临时设置当前用户，以便对象将以
	 * 正确的所有权创建。
	 *
	 * （该设置将在此例程结束时恢复，或者在发生错误时，
	 * 事务中止将清理。
	 */
	if (fc_saved_uid != fc_owner_uid)
		SetUserIdAndSecContext(fc_owner_uid,
							   fc_save_sec_context | SECURITY_LOCAL_USERID_CHANGE);

	/* 创建模式的命名空间 */
	fc_namespaceId = NamespaceCreate(fc_schemaName, fc_owner_uid, false);

	/* 增加命令计数器以使命名空间可见 */
	CommandCounterIncrement();

	/*
	 * 在当前搜索路径前面添加新模式。
	 *
	 * 我们使用等效的函数 SET 选项，以允许设置
	 * 在模式创建期间持续存在。guc.c 也会处理在错误时撤销设置。
	 */
	fc_save_nestlevel = NewGUCNestLevel();

	initStringInfo(&fc_pathbuf);
	appendStringInfoString(&fc_pathbuf, quote_identifier(fc_schemaName));

	while (scanner_isspace(*fc_nsp))
		fc_nsp++;

	if (*fc_nsp != '\0')
		appendStringInfo(&fc_pathbuf, ", %s", fc_nsp);

	(void) set_config_option("search_path", fc_pathbuf.data,
							 PGC_USERSET, PGC_S_SESSION,
							 GUC_ACTION_SAVE, true, 0, false);

	/*
	 * 向可能感兴趣的事件触发器报告新模式。注意，我们
	 * 必须在这里执行此操作，而不是在 ProcessUtilitySlow 中，因为如果不这样做，
	 * 下面创建的对象在模式之前被报告，这是错误的。
	 */
	ObjectAddressSet(fc_address, NamespaceRelationId, fc_namespaceId);
	EventTriggerCollectSimpleCommand(fc_address, InvalidObjectAddress,
									 (Node *) fc_stmt);

	/*
	 * 检查嵌入在 CREATE SCHEMA 命令中的命令列表，并
	 * 将它们重组为顺序可执行的顺序，不带前向
	 * 引用。注意，结果仍然是原始解析树的列表 ---
	 * 一般来说，我们不能在执行之前的命令之前对一个语句进行解析分析。
	 */
	fc_parsetree_list = transformCreateSchemaStmtElements(fc_stmt->schemaElts,
													   fc_schemaName);

	/*
	 * 执行 CREATE SCHEMA 中包含的每个命令。由于语法
	 * 仅允许实用程序命令在 CREATE SCHEMA 中，因此没有必要将它们传递
	 * 通过 parse_analyze_*() 或重写器；我们可以直接将它们交给 ProcessUtility。
	 */
	foreach(fc_parsetree_item, fc_parsetree_list)
	{
		Node	   *fc_stmt = (Node *) lfirst(fc_parsetree_item);
		PlannedStmt *fc_wrapper;

		/* 需要创建一个包装 PlannedStmt */
		fc_wrapper = makeNode(PlannedStmt);
		fc_wrapper->commandType = CMD_UTILITY;
		fc_wrapper->canSetTag = false;
		fc_wrapper->utilityStmt = fc_stmt;
		fc_wrapper->stmt_location = fc_stmt_location;
		fc_wrapper->stmt_len = fc_stmt_len;

		/* 执行此步骤 */
		ProcessUtility(fc_wrapper,
					   fc_queryString,
					   false,
					   PROCESS_UTILITY_SUBCOMMAND,
					   NULL,
					   NULL,
					   None_Receiver,
					   NULL);

		/* 确保后续步骤可以看到这里创建的对象 */
		CommandCounterIncrement();
	}

	/*
	 * 恢复我们上面设置的 GUC 变量 search_path。
	 */
	AtEOXact_GUC(true, fc_save_nestlevel);

	/* 重置当前用户和安全上下文 */
	SetUserIdAndSecContext(fc_saved_uid, fc_save_sec_context);

	return fc_namespaceId;
}


/*
 * 重命名模式
 */
ObjectAddress RenameSchema(const char *fc_oldname, const char *fc_newname)
{
	Oid			fc_nspOid;
	HeapTuple	fc_tup;
	Relation	fc_rel;
	AclResult	fc_aclresult;
	ObjectAddress fc_address;
	Form_pg_namespace fc_nspform;

	fc_rel = table_open(NamespaceRelationId, RowExclusiveLock);

	fc_tup = SearchSysCacheCopy1(NAMESPACENAME, CStringGetDatum(fc_oldname));
	if (!HeapTupleIsValid(fc_tup))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_SCHEMA),
				 errmsg("schema \"%s\" does not exist", fc_oldname)));

	fc_nspform = (Form_pg_namespace) GETSTRUCT(fc_tup);
	fc_nspOid = fc_nspform->oid;

	/* 确保新名称不存在 */
	if (OidIsValid(get_namespace_oid(fc_newname, true)))
		ereport(ERROR,
				(errcode(ERRCODE_DUPLICATE_SCHEMA),
				 errmsg("schema \"%s\" already exists", fc_newname)));

	/* 必须是所有者 */
	if (!pg_namespace_ownercheck(fc_nspOid, GetUserId()))
		aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SCHEMA,
					   fc_oldname);

	/* 必须对数据库具有 CREATE 权限 */
	fc_aclresult = pg_database_aclcheck(MyDatabaseId, GetUserId(), ACL_CREATE);
	if (fc_aclresult != ACLCHECK_OK)
		aclcheck_error(fc_aclresult, OBJECT_DATABASE,
					   get_database_name(MyDatabaseId));

	if (!allowSystemTableMods && IsReservedName(fc_newname))
		ereport(ERROR,
				(errcode(ERRCODE_RESERVED_NAME),
				 errmsg("unacceptable schema name \"%s\"", fc_newname),
				 errdetail("The prefix \"pg_\" is reserved for system schemas.")));

	/* 重命名 */
	namestrcpy(&fc_nspform->nspname, fc_newname);
	CatalogTupleUpdate(fc_rel, &fc_tup->t_self, fc_tup);

	InvokeObjectPostAlterHook(NamespaceRelationId, fc_nspOid, 0);

	ObjectAddressSet(fc_address, NamespaceRelationId, fc_nspOid);

	table_close(fc_rel, NoLock);
	heap_freetuple(fc_tup);

	return fc_address;
}

void AlterSchemaOwner_oid(Oid fc_oid, Oid fc_newOwnerId)
{
	HeapTuple	fc_tup;
	Relation	fc_rel;

	fc_rel = table_open(NamespaceRelationId, RowExclusiveLock);

	fc_tup = SearchSysCache1(NAMESPACEOID, ObjectIdGetDatum(fc_oid));
	if (!HeapTupleIsValid(fc_tup))
		elog(ERROR, "cache lookup failed for schema %u", fc_oid);

	fc_AlterSchemaOwner_internal(fc_tup, fc_rel, fc_newOwnerId);

	ReleaseSysCache(fc_tup);

	table_close(fc_rel, RowExclusiveLock);
}


/*
 * 更改模式所有者
 */
ObjectAddress AlterSchemaOwner(const char *fc_name, Oid fc_newOwnerId)
{
	Oid			fc_nspOid;
	HeapTuple	fc_tup;
	Relation	fc_rel;
	ObjectAddress fc_address;
	Form_pg_namespace fc_nspform;

	fc_rel = table_open(NamespaceRelationId, RowExclusiveLock);

	fc_tup = SearchSysCache1(NAMESPACENAME, CStringGetDatum(fc_name));
	if (!HeapTupleIsValid(fc_tup))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_SCHEMA),
				 errmsg("schema \"%s\" does not exist", fc_name)));

	fc_nspform = (Form_pg_namespace) GETSTRUCT(fc_tup);
	fc_nspOid = fc_nspform->oid;

	fc_AlterSchemaOwner_internal(fc_tup, fc_rel, fc_newOwnerId);

	ObjectAddressSet(fc_address, NamespaceRelationId, fc_nspOid);

	ReleaseSysCache(fc_tup);

	table_close(fc_rel, RowExclusiveLock);

	return fc_address;
}

static void fc_AlterSchemaOwner_internal(HeapTuple fc_tup, Relation fc_rel, Oid fc_newOwnerId)
{
	Form_pg_namespace fc_nspForm;

	Assert(fc_tup->t_tableOid == NamespaceRelationId);
	Assert(RelationGetRelid(fc_rel) == NamespaceRelationId);

	fc_nspForm = (Form_pg_namespace) GETSTRUCT(fc_tup);

	/*
	 * 如果新所有者与现有所有者相同，则视为命令已成功。
	 * 这是为了转储恢复的目的。
	 */
	if (fc_nspForm->nspowner != fc_newOwnerId)
	{
		Datum		fc_repl_val[Natts_pg_namespace];
		bool		fc_repl_null[Natts_pg_namespace];
		bool		fc_repl_repl[Natts_pg_namespace];
		Acl		   *fc_newAcl;
		Datum		fc_aclDatum;
		bool		fc_isNull;
		HeapTuple	fc_newtuple;
		AclResult	fc_aclresult;

		/* 否则，必须是现有对象的所有者 */
		if (!pg_namespace_ownercheck(fc_nspForm->oid, GetUserId()))
			aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SCHEMA,
						   NameStr(fc_nspForm->nspname));

		/* 必须能够成为新所有者 */
		check_is_member_of_role(GetUserId(), fc_newOwnerId);

		/*
		 * 必须有创建模式的权限
		 *
		 * 注意：这与其他更改所有者检查不同，因为当前用户被检查以
		 * 确认其创建权限，而不是目标所有者。这与模式的 CREATE 情况一致。
		 * 因为超级用户总是会拥有这一权利，我们不需要为他们设置特殊情况。
		 */
		fc_aclresult = pg_database_aclcheck(MyDatabaseId, GetUserId(),
										 ACL_CREATE);
		if (fc_aclresult != ACLCHECK_OK)
			aclcheck_error(fc_aclresult, OBJECT_DATABASE,
						   get_database_name(MyDatabaseId));

		memset(fc_repl_null, false, sizeof(fc_repl_null));
		memset(fc_repl_repl, false, sizeof(fc_repl_repl));

		fc_repl_repl[Anum_pg_namespace_nspowner - 1] = true;
		fc_repl_val[Anum_pg_namespace_nspowner - 1] = ObjectIdGetDatum(fc_newOwnerId);

		/*
		 * 确定新所有者的修改后 ACL。这仅在 ACL 非空时必要。
		 */
		fc_aclDatum = SysCacheGetAttr(NAMESPACENAME, fc_tup,
								   Anum_pg_namespace_nspacl,
								   &fc_isNull);
		if (!fc_isNull)
		{
			fc_newAcl = aclnewowner(DatumGetAclP(fc_aclDatum),
								 fc_nspForm->nspowner, fc_newOwnerId);
			fc_repl_repl[Anum_pg_namespace_nspacl - 1] = true;
			fc_repl_val[Anum_pg_namespace_nspacl - 1] = PointerGetDatum(fc_newAcl);
		}

		fc_newtuple = heap_modify_tuple(fc_tup, RelationGetDescr(fc_rel), fc_repl_val, fc_repl_null, fc_repl_repl);

		CatalogTupleUpdate(fc_rel, &fc_newtuple->t_self, fc_newtuple);

		heap_freetuple(fc_newtuple);

		/* 更新所有者依赖关系引用 */
		changeDependencyOnOwner(NamespaceRelationId, fc_nspForm->oid,
								fc_newOwnerId);
	}

	InvokeObjectPostAlterHook(NamespaceRelationId,
							  fc_nspForm->oid, 0);
}
